#include "app_pool.h"
#include  <mqueue.h>
#include  <pthread.h>
#include <stdio.h>
#include "log.h"

static mqd_t mqd; //消息队列描述符
static pthread_t  threads[APP_POOL_THREAD_NUM];

/**
 * @brief 线程执行的函数
 */
void *app_pool_thread_func(void *arg)
{
    //1 线程主动从队列中拉取任务
    task_t task ;
    while(1)
    {   
        int size=  mq_receive(mqd,(char *) &task,sizeof(task),0); 
        if(size>0)
        {   
            //2 拉取到数据后进行处理
            task.task_func(task.task_arg);
        }    
    }
   
    return NULL;

}

/**
 * @brief 初始化线程池 
 * 
 * 1  队列
 * 2  线程池 
 */
int app_pool_init()
{  //1 创建队列 
    mq_unlink(APP_POOL_QUEUE_NAME);//否则 如果已有队列，那么队列不会重新创建，新的参数也不会被设置
    struct mq_attr attr;
    attr.mq_maxmsg=APP_POOL_QUEUE_MAXMSG;
    attr.mq_msgsize=sizeof(task_t);
    
    mqd=mq_open(APP_POOL_QUEUE_NAME,O_CREAT|O_RDWR,0666,&attr);
    if(mqd==-1)
    {
        perror("mq_open");
        return -1;
    }
    log_info("线程池队列创建完成");

   //2 创建线程池
   for (size_t i = 0; i < APP_POOL_THREAD_NUM; i++)
   {
       pthread_create(&threads[i],NULL,app_pool_thread_func,NULL);   
       log_info("创建一个线程");
   }
   
   return 0;

}



/**
 * @brief  向线程池提交任务
 */
int app_pool_submit_task(void *(*func)(void *), void *arg)
{
    task_t task=
    {
        .task_func=func,
        .task_arg=arg  
    };
    int ret= mq_send(mqd,(char *) &task,sizeof(task),0);
    if(ret==-1)
    {
        perror("提交任务队列失败");
        return -1;
    }
    return 0;

}

/**
 * @brief 销毁线程池
 */
void app_pool_destroy()
{
    //1 线程池 结束掉 
    for (size_t i = 0; i < APP_POOL_THREAD_NUM; i++)
    {
         pthread_cancel(threads[i]);
    }
    //等待所有线程结束
    for (size_t i = 0; i < APP_POOL_THREAD_NUM; i++) 
    {
         pthread_join(threads[i],NULL);
    }
    log_info("已结束所有线程");
    //2 队列关闭 
    mq_close(mqd);
    //3 队列删除
    mq_unlink(APP_POOL_QUEUE_NAME);
    log_info("已销毁队列");

}